﻿using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using Confluent.Kafka;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using NetMicro.Core.Extensions;

namespace NetMicro.EventBus.Kafka
{
    public class KafkaConnection : IKafkaConnection
    {

        private readonly EventBusKafkaOptions _options;
        private const int DefaultPoolSize = 15;
        private readonly ConcurrentQueue<IProducer<string, byte[]>> _producerPool;
        private int _pCount;
        private int _maxSize;
        private ConcurrentDictionary<string, IConsumer<string, byte[]>> Consumers { get; } =
            new ConcurrentDictionary<string, IConsumer<string, byte[]>>();

        public KafkaConnection(ILogger<KafkaConnection> logger, IOptions<EventBusKafkaOptions> options)
        {
            _options = options.Value;
            _producerPool = new ConcurrentQueue<IProducer<string, byte[]>>();
            _maxSize = DefaultPoolSize;

            logger.LogDebug("Kafka servers: {0}", _options.Properties.ToJson());
        }



        public IProducer<string, byte[]> RentProducer()
        {
            if (_producerPool.TryDequeue(out var producer))
            {
                Interlocked.Decrement(ref _pCount);

                return producer;
            }
            var config = new ProducerConfig(new Dictionary<string, string>(_options.Properties))
            {
                QueueBufferingMaxMessages = 10,
                MessageTimeoutMs = 5000,
                RequestTimeoutMs = 3000
            };

            producer = new ProducerBuilder<string, byte[]>(config).Build();
            return producer;
        }

        public bool Return(IProducer<string, byte[]> producer)
        {
            if (Interlocked.Increment(ref _pCount) <= _maxSize)
            {
                _producerPool.Enqueue(producer);

                return true;
            }

            Interlocked.Decrement(ref _pCount);

            return false;
        }

        public void Dispose()
        {
            _maxSize = 0;

            while (_producerPool.TryDequeue(out var context))
            {
                context.Dispose();

            }
        }
        public IConsumer<string, byte[]> CreateConsumer(string eventHandlerName)
        {
            var dic = _options.Properties;
            dic["group.id"] = eventHandlerName;
            // 允许自动创建topic
            dic.TryAdd("allow.auto.create.topics", "true");
            // 禁止自动保存offset
            dic.TryAdd("enable.auto.offset.store", "false");
            // 启用自动提交
            dic.TryAdd("enable.auto.commit", "false");
            dic.TryAdd("auto.offset.reset", "earliest");
            return Consumers.GetOrAdd(eventHandlerName, r =>
                new ConsumerBuilder<string, byte[]>(dic).Build()
            );
        }
    }
}